package com.atguigu.gmall.realtime.joindemo;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/12/6 10:12
 */
public class LookupJoinDemo extends BaseSQLApp {
    public static void main(String[] args) {
        new LookupJoinDemo().init(
            6666,
            2,
            "LookupJoinDemo"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        
        tEnv.executeSql("create table t1(" +
                            " id string, " +
                            " pt as proctime() " +
                            ")" + SQLUtil.getKafkaSourceDDL("t1", "atguigu", "csv"));
    
        
        tEnv.executeSql("create temporary table base_dic (" +
                            "  dic_code string," +
                            "  dic_name string " +
                            ") WITH (" +
                            "  'connector' = 'jdbc'," +
                            "  'url' = 'jdbc:mysql://hadoop162:3306/gmall2022?useSSL=false'," +
                            "  'table-name' = 'base_dic', " +
                            "  'username' = 'root', " +
                            "  'lookup.cache.ttl' = '30 s', " +  // 用来设置维度表的数据的ttl 时间
                            "  'lookup.cache.max-rows' = '10', " +  // 缓存的最大行数
                            "  'password' = 'aaaaaa' " +
                            ")");
    
        tEnv.sqlQuery("select " +
                          " t1.id, " +
                          " dic.dic_name " +
                          "from t1 " +
                          "join base_dic for system_time as of t1.pt as dic " +
                          "on t1.id=dic.dic_code")
            .execute()
            .print();
        
        
        
    
     
    }
}
/*
look upjoin:
    作用: 就是补充维度信息
        实时表 join 维度表
        
    左表: 有处理时间
    右表: 来源与外部系统(jdbc Source connector) mysql

"  'lookup.cache.ttl' = '30 s', " +  // 用来设置维度表的数据的ttl 时间
 "  'lookup.cache.max-rows' = '10', " +  // 缓存的最大行数
 
 1. 提高读的效率 2.减少对 mysql 的眼里
 
    坏处: 结果可能不准
 */